package com.exsun.gateway.jt809.kafka;

import com.exsun.gateway.jt809.common.DevInfoService;
import com.exsun.gateway.jt809.common.LinkType;
import com.exsun.gateway.jt809.data.PushData;
import com.exsun.gateway.jt809.common.LinkMgr;
import exsun.jt808.data.kafka_data.KafkaJT808Data;
import exsun.jt808.data.upload_data.GpsData;
import exsun.jt808.data.upload_data.Jt808EntityBase;
import exsun.jt808.data.util.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

public class JT809KafkaHandleBase<T extends Jt808EntityBase> {
    private Logger logger = LoggerFactory.getLogger(JT809KafkaHandleBase.class);
    @Autowired
    private DevInfoService devInfoService;

    public List<PushData> done(List<ConsumerRecord<?, ?>> records, Type type, Function<T, List<PushData>> action) throws Exception{
//        Date dt = new Date();
        List<PushData> pushs = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : records) {
            //任意链路都不通，则不处理数据
            if(!LinkMgr.getLinkStatus())return null;
            String msg = record.value().toString();
            KafkaJT808Data<T> kafkaJT808ReviveData = GsonUtils.getObjectFromJson(msg, type);
            T obj = kafkaJT808ReviveData.getObject();
            if(obj == null||!devInfoService.bSendTo809(obj.dvo))continue;
            List<PushData> temp = action.apply(obj);
            if(temp == null||temp.size() ==0) continue;
            pushs.addAll(temp);
        }
        boolean result = LinkMgr.push2FitChannel(pushs, LinkType.MASTER);
        if(!result)return null;

//        long span = new Date().getTime() - dt.getTime();
//        logger.info("处理[{}]数据，花费时间{}ms",type.getTypeName(), span);
        return pushs;
    }
    public List<PushData> done1(List<ConsumerRecord<?, ?>> records, Type type, Function<T, List<PushData>> action) throws Exception{
//        Date dt = new Date();
        List<PushData> pushs = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : records) {
            String msg = record.value().toString();
            GpsData object = GsonUtils.getObjectFromJson(msg, type);
            T obj = (T) object;
            if(obj == null||!devInfoService.bSendTo809(obj.dvo))continue;
            List<PushData> temp = action.apply(obj);
            if(temp == null||temp.size() ==0) continue;
            pushs.addAll(temp);
        }
        boolean result = LinkMgr.push2FitChannel(pushs, LinkType.MASTER);
        if(!result)return null;

//        long span = new Date().getTime() - dt.getTime();
//        logger.info("处理[{}]数据，花费时间{}ms",type.getTypeName(), span);
        return pushs;
    }
}
